package com.kingtao.rpc;

import com.kingtao.entry.LogEntry;
import com.kingtao.entry.Node;
import com.kingtao.entry.NodeGroup;
import com.kingtao.entry.Peer;
import com.kingtao.enums.NodeStatusEnums;
import com.kingtao.message.*;
import com.kingtao.protocol.MessageCodec;
import com.kingtao.protocol.ProtocolFrameDecoder;
import com.kingtao.service.Impl.LogServiceImpl;
import com.kingtao.service.Impl.NodeGroupServiceImpl;
import com.kingtao.util.CoreData;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @Author: kingtao
 * @Date: 2023-08-25-15:21
 * @Description:
 */
@Slf4j
public class server {

    public static List<LogEntry> logEntries = null;
    public static void main(String[] args) {
        List<Node> peers = new ArrayList<>();
        NodeGroupServiceImpl groupService = new NodeGroupServiceImpl();

        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        //  各种handler
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);

        MessageCodec CODEC_HANDLER = new MessageCodec();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast(new ProtocolFrameDecoder());
                            channel.pipeline().addLast(LOGGING_HANDLER);
                            channel.pipeline().addLast(CODEC_HANDLER);
                            //  初始化server
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<ServerInitRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, ServerInitRequest msg) throws Exception {
                                    Integer num = msg.getNum();
                                    //  默认最少5个节点
                                    if (num < CoreData.NODE_NUM) {
                                        num = CoreData.GROUP_NUM;
                                    }
                                    for (int i = 0; i < num; i++) {
                                        Peer peer = new Peer(CoreData.NODE_ADDRESS, CoreData.NODE_INIT_PORT + i);
                                        Node node = new Node(i, peer);
                                        peers.add(node);
                                    }
                                    NodeGroup group = new NodeGroup(peers);
                                    groupService.setNodeGroup(group);
                                    groupService.setLeader();
                                    ServerInitResponse response = new ServerInitResponse(true, num);
                                    ctx.writeAndFlush(response);
                                    ctx.writeAndFlush(groupService.getNodeGroup().toString());
                                }
                            });
                            //  筛选leader
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<SelectLeaderRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, SelectLeaderRequest msg) throws Exception {
                                    SelectLeaderResponse response;
                                    int id;
                                    if (msg.getId() != 0) {
                                        id = msg.getId();
                                    } else {
                                        id = 0;
                                    }
                                    //  leader 依然为 leader
                                    if (id == 0) {
                                        groupService.setLeader();
                                        response = new SelectLeaderResponse(groupService.getNodeGroup().getLeaderId(), true);
                                    } else if (groupService.getNodeGroup().getNodeByNodeId(id).getStatus() == NodeStatusEnums.leader) {
                                        response = new SelectLeaderResponse(id, true);
                                    } else {
                                        //  leader 不是 leader
                                        groupService.setLeader();
                                        response = new SelectLeaderResponse(groupService.getNodeGroup().getLeaderId(), true);
                                    }
                                    ctx.writeAndFlush(response);
                                }
                            });
                            //  筛选 secondLeader
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<SelectSecondLeaderRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, SelectSecondLeaderRequest msg) throws Exception {
                                    SelectSecondLeaderResponse response;
                                    int id;
                                    if (msg.getId() != 0) {
                                        id = msg.getId();
                                    } else {
                                        id = 0;
                                    }
                                    if (id == 0) {
                                        groupService.setFollowerLeader();
                                        response = new SelectSecondLeaderResponse(groupService.getNodeGroup().getSecondLeaderId(), true);
                                    } else if (groupService.getNodeGroup().getNodeByNodeId(id).getStatus() == NodeStatusEnums.secondLeader) {
                                        response = new SelectSecondLeaderResponse(id, true);
                                    } else {
                                        groupService.setSecondLeader();
                                        response = new SelectSecondLeaderResponse(groupService.getNodeGroup().getSecondLeaderId(), true);
                                    }
                                    ctx.writeAndFlush(response);
                                }
                            });
                            //  筛选 followerLeader
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<SelectFollowerLeaderRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, SelectFollowerLeaderRequest msg) throws Exception {
                                    groupService.setCreateGroup(true);
                                    SelectFollowerLeaderResponse response;
                                    if (msg.getId() == 0) {
                                        groupService.setFollowerLeader();
                                        response = new SelectFollowerLeaderResponse(true);
                                    } else {
                                        //  任然为 follower
                                        if (groupService.getNodeGroup().getNodeByNodeId(msg.getId()).getStatus() == NodeStatusEnums.followerLeader) {
                                            log.debug("Node {} 依然为组长节点", msg.getId());
                                        } else {
                                            groupService.setFollowerLeaderById(msg.getId());
                                        }
                                        response = new SelectFollowerLeaderResponse(true);
                                    }
                                    ctx.writeAndFlush(response);
                                    ctx.writeAndFlush(groupService.getNodeGroup().toString());
                                }
                            });
                            //  添加node
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<AppendNodeRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, AppendNodeRequest msg) throws Exception {
                                    AppendNodeResponse response;
                                    if (msg.getNum() > 0) {
                                        int size = groupService.getNodeGroup().getNodes().size();
                                        for (int i = 0; i < msg.getNum(); i++) {
                                            Node node = new Node(i + size, new Peer("127.0.0.1", i + size + CoreData.NODE_INIT_PORT));
                                            groupService.appendNode(node);
                                        }
                                        response = new AppendNodeResponse(msg.getNum(), true);
                                    } else {
                                        response = new AppendNodeResponse(msg.getNum(), false);
                                    }
                                    ctx.writeAndFlush(response);
                                    ctx.writeAndFlush(groupService.getNodeGroup().toString());
                                }
                            });
                            //  集群添加客户端请求
                            /**
                             * 1、leader写入 验证自己
                             * 2、secondLeader写入，验证自己，验证leader
                             * 3、写入followerLeader，验证自己
                             * 4、写入follower，验证自己
                             * 5、写入成功（写入比例大于0.5）
                             */
                            channel.pipeline().addLast(new SimpleChannelInboundHandler<ClientRequest>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, ClientRequest msg) throws Exception {
                                    String str = msg.getStr();
                                    LogEntry logEntry = new LogEntry(str);
                                    int index = 0;
                                    Node leader = groupService.getLeader();
                                    Node secondLeader = groupService.getSecondLeader();

                                    if (leader.getStatus() == NodeStatusEnums.leader) {
                                        LogServiceImpl logService = new LogServiceImpl(leader);
                                        //  leader 写入
                                        logService.writeLogEntry(logEntry);
                                        if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str) ) {
                                            log.debug("leader 写入成功");
                                            groupService.rewardNode(leader.getId());
                                            //  新增后 leader 的 lastIndex 自增
                                            index = logService.getLastIndex()-1;
                                            logEntries = logService.getAllLogEntry();
                                            log.debug("{}", logEntries);
                                            logService.destroy();
                                        }else {
                                            //  leader作恶
                                            logService.destroy();
                                            boolean flag = true;
                                            while(flag) {
                                                log.debug("leader 不诚实, 重新筛选");
                                                //  leader换人了
                                                groupService.punishNode(leader.getId());
                                                LogServiceImpl newLogService = new LogServiceImpl(groupService.getLeader());
                                                newLogService.writeLogEntry(logEntry);
                                                groupService.rewardNode(groupService.getLeader().getId());
                                                index = newLogService.getLastIndex()-1;
                                                if (newLogService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                    flag = false;
                                                    logEntries = newLogService.getAllLogEntry();
                                                    log.debug("新的Leader 写入成功");
                                                }
                                                newLogService.destroy();
                                            }
                                        }
                                    }

                                    if (secondLeader.getStatus() == NodeStatusEnums.secondLeader) {
                                        LogServiceImpl logService = new LogServiceImpl(secondLeader);
                                        //  secondLeader同步 leader
                                        if (index != logService.getLastIndex()) {
                                            logService.updateAllLog(logEntries);
                                        }
                                        logService.writeLogEntry(logEntry);
                                        if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str) ) {
                                            log.debug("secondLeader 写入成功");
                                            groupService.rewardNode(secondLeader.getId());
                                            logService.destroy();
                                        }else {
                                            //  leader作恶
                                            logService.destroy();
                                            boolean flag = true;
                                            while(flag) {
                                                log.debug("secondLeader 不诚实, 重新筛选");
                                                //  leader换人了
                                                groupService.punishNode(secondLeader.getId());
                                                LogServiceImpl newLogService = new LogServiceImpl(groupService.getSecondLeader());
                                                newLogService.updateLastIndexKey(index);
                                                newLogService.writeLogEntry(logEntry);
                                                if (newLogService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                    flag = false;
                                                    groupService.rewardNode(groupService.getSecondLeader().getId());
                                                    log.debug("新的secondLeader 写入成功");
                                                }
                                                newLogService.destroy();
                                            }
                                        }
                                    }

                                    if (groupService.isCreateGroup() == true) {
                                        //  分组了
                                        Map<Integer, Integer> followerLeaderIdMap = groupService.getNodeGroup().getFollowerLeaderIdMap();
                                        Node node;
                                        for (int i=0; i<followerLeaderIdMap.size(); i++) {
                                            //  获取组长
                                            node = groupService.getNodeGroup().getNodeByNodeId(followerLeaderIdMap.get(i));
                                            if (node.getStatus() == NodeStatusEnums.followerLeader) {
                                                //  同步 secondLeader
                                                LogServiceImpl logService = new LogServiceImpl(node);
                                                if (index < logService.getLastIndex()) {
                                                    logService.updateAllLog(logEntries);
                                                }
                                                if (index >= logService.getLastIndex()) {
                                                    logService.writeLogEntry(logEntry);
                                                    groupService.rewardNode(node.getId());
                                                    if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str) ) {
                                                        log.debug("followerLeader {} 写入成功", node.getId());
                                                        logService.destroy();
                                                    }else {
                                                        //  followerLeader 作恶
                                                        boolean flag = false;
                                                        while(flag) {
                                                            log.debug("followerLeader 不诚实");
                                                            logService.destroy();
                                                            groupService.punishNode(node.getId());
                                                            node = groupService.getNodeGroup().getNodeByNodeId(followerLeaderIdMap.get(i));
                                                            LogServiceImpl newLogService = new LogServiceImpl(node);
                                                            newLogService.updateLastIndexKey(index);
                                                            newLogService.writeLogEntry(logEntry);
                                                            if (newLogService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                                flag = true;
                                                                log.debug("followerLeader {} 写入成功", node.getId());
                                                                newLogService.destroy();
                                                            }
                                                        }
                                                    }
                                                }
                                            }
                                            //  followerLeader 启动写入线程,将数据写到 follower
                                            int finalI = i;
                                            int finalIndex = index;
                                            List<LogEntry> finalLogEntries = logEntries;
                                            new Thread(() -> {
                                                for (int j = finalI-1; j < groupService.getNodeGroup().getNodes().size(); j = j + CoreData.GROUP_NUM) {
                                                    Node follower = groupService.getNodeGroup().getNodeByNodeId(j+1);
                                                    if (follower == null) {
                                                        continue;
                                                    }
                                                    if (follower.getStatus() == NodeStatusEnums.follower) {
                                                        LogServiceImpl logService = new LogServiceImpl(follower);
                                                        if (finalIndex < logService.getLastIndex()) {
                                                            logService.updateAllLog(finalLogEntries);
                                                        }
                                                        logService.updateLastIndexKey(finalIndex);
                                                        logService.writeLogEntry(logEntry);

                                                        if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                            groupService.rewardNode(follower.getId());
                                                            log.debug("follower {} 写入成功", follower.getId());
                                                        } else {
                                                            //  follower 作恶
                                                            boolean flag = false;
                                                            while (flag) {
                                                                log.debug("follower 不诚实");
                                                                groupService.punishNode(follower.getId());
                                                                logService.updateLastIndexKey(finalIndex);
                                                                logService.writeLogEntry(logEntry);
                                                                if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                                    flag = true;
                                                                    groupService.rewardNode(follower.getId());
                                                                    log.debug("follower {} 写入成功", follower.getId());
                                                                }
                                                            }
                                                        }
                                                        logService.destroy();
                                                    }
                                                }
                                            }, "followerLeader-"+finalI).start();
                                        }
                                    }else {
                                        //  没有分组
                                        for (Node node : groupService.getNodeGroup().getAllFollower()) {
                                            //  同步 secondLeader
                                            LogServiceImpl logService = new LogServiceImpl(node);
                                            if (index < logService.getLastIndex()) {
                                                logService.updateAllLog(logEntries);
                                            }
                                            if (index >= logService.getLastIndex()) {
                                                logService.writeLogEntry(logEntry);
                                                groupService.rewardNode(node.getId());
                                                if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str) ) {
                                                    log.debug("node {} 写入成功", node.getId());
                                                }else {
                                                    //  node 作恶
                                                    boolean flag = false;
                                                    while(flag) {
                                                        log.debug("node 不诚实");
                                                        groupService.punishNode(node.getId());
                                                        logService.updateLastIndexKey(index);
                                                        logService.writeLogEntry(logEntry);
                                                        if (logService.getValueFromIndexByKey(logService.getLastIndex(), 1).equals(str)) {
                                                            flag = true;
                                                            log.debug("node {} 写入成功", node.getId());
                                                        }
                                                    }
                                                }
                                            }
                                            logService.destroy();
                                        }
                                    }

                                    ClientResponse response = new ClientResponse("写入成功", true);
                                    ctx.writeAndFlush(response);
                                    ctx.writeAndFlush(groupService.getNodeGroup().toString());
                                }
                            });
                            //  心跳检测
                            channel.pipeline().addLast(new IdleStateHandler(1, 0, 0));
                            //  入栈出栈 处理器
                            channel.pipeline().addLast(new ChannelDuplexHandler() {
                                //  用来触发特殊事件
                                @Override
                                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                                    HeartBeatResponse response;
                                    IdleStateEvent event = (IdleStateEvent) evt;
                                    if (event.state() == IdleState.READER_IDLE) {
                                        if (groupService.isCreateGroup() == false) {
                                            log.debug("leader id: {},secondLeader id: {}", groupService.getLeader().getId(), groupService.getSecondLeader().getId());
                                        } else {
                                            log.debug("leader id: {},secondLeader id: {}, followerLeader id: {}",
                                                    groupService.getLeader().getId(), groupService.getSecondLeader().getId(), groupService.getNodeGroup().getFollowerLeaderIdMap());
                                        }
                                        if (groupService.getLeader().getStatus() != NodeStatusEnums.leader
                                                && groupService.getSecondLeader().getStatus() != NodeStatusEnums.secondLeader) {
                                            response = new HeartBeatResponse(false, false);
                                            log.debug("leader与secondLeader出现问题,重新选择");
                                            ctx.writeAndFlush(response);
                                        } else if (groupService.getSecondLeader().getStatus() != NodeStatusEnums.secondLeader) {
                                            response = new HeartBeatResponse(true, false);
                                            log.debug("secondLeader出现问题,重新选择");
                                            ctx.writeAndFlush(response);
                                        }
                                        else if (groupService.getLeader().getStatus() != NodeStatusEnums.leader) {
                                            response = new HeartBeatResponse(false, true);
                                            log.debug("leader出现问题,重新选择");
                                            ctx.writeAndFlush(response);
                                        }
                                    }
                                }
                            });
                        }
                    });
            ChannelFuture future = serverBootstrap.bind(CoreData.NODE_INIT_PORT);
            future.sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.debug("{}", e.getMessage());
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}
